package com.rrd.dw.canal.client;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
import com.rrd.dw.canal.Utils.PropertiesUtil;
import com.rrd.dw.canal.sink.KafkaSink;

public class AccountCannlClient extends AbstractCanalClient{
	public AccountCannlClient(String destination) {
		super(destination);
	}
	public AccountCannlClient(String destination, CanalConnector connector) {
		super(destination,connector);
	}
	
	public static void main(String args[]) {
        String destination = "account";
        String topic ="account.loan";
        String subscribe="";
        if(args!=null){
        	if(args.length>=1){
        		destination=args[0].trim();
        	}
        	if(args.length>=2){
        		topic=args[1];
        	}
        	if(args.length>=3){
        		subscribe=args[2];
        	}
        	
        }
        // 基于zookeeper动态获取canal server的地址，建立链接，其中一台server发生crash，可以支持failover
        String zk=PropertiesUtil.getVal("canal.zookeeper");
        
        logger.info("destination=="+destination);
        logger.info("topic=="+topic);
        logger.info("zookeeper=="+zk);
        
        CanalConnector connector = CanalConnectors.newClusterConnector(zk,destination, "", "");
        String ip = "10.10.15.65";
//        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111), destination, "", "");
        connector.connect();
		connector.subscribe(subscribe);
        final AccountCannlClient client = new AccountCannlClient(destination);
        final KafkaSink sink = new KafkaSink(topic);
        client.setConnector(connector);
        client.setSink(sink);
        client.start();

        Runtime.getRuntime().addShutdownHook(new Thread() {

            public void run() {
                try {
                    logger.info("## stop the canal client");
                    client.stop();
                    sink.closeProducer();
                } catch (Throwable e) {
                    logger.warn("##something goes wrong when stopping canal:", e);
                } finally {
                    logger.info("## canal client is down.");
                }
            }
           
        });
    }
	/**
	 * 解析message
	 * @param message
	 */
	protected void parseMessage(Message message) {
		List<Entry> entrys = message.getEntries();
		for (Entry entry : entrys) {
			if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
					|| entry.getEntryType() == EntryType.TRANSACTIONEND) {
				continue;
			} else if (entry.getEntryType() == EntryType.ROWDATA) {
				RowChange rowChage = null;
				try {
					rowChage = RowChange.parseFrom(entry.getStoreValue());
				} catch (Exception e) {
					throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
				}
				EventType eventType = rowChage.getEventType();
					if(eventType==EventType.INSERT){
						 rowDataToString(rowChage);
					}
			}
		}
	}
	/**
	 * 解析rowdata，然后写入到存储
	 * @param rowData
	 * @param type
	 */
	private void rowDataToString(RowChange rowChanage){
		StringBuilder sb = new StringBuilder(500);
		List<Column> columnList = null;
		 Map<String,String> map = new HashMap<String, String>();
//		sb.append(rowChanage.getEventType());
		 map.put("oprType", rowChanage.getEventType().toString());
		for (RowData rowData : rowChanage.getRowDatasList()) {
			columnList=rowData.getAfterColumnsList();
			for(Column c : columnList){
//				sb.append(CRT001).append(c.getValue()==null?"":c.getValue());
				map.put(c.getName(), c.getValue());
			}
		}
		
		if(this.sink==null){
			printColumn(columnList);
		}else{
			printColumn(columnList);
//			this.sink.write(columnList.get(2).getValue(),sb.toString().replace("\n", ""));
			this.sink.write(map.get("loan_key"),map.get("channel_code")+","+map.get("start_date")+","+map.get("amount")+","+map.get("early_service_fee"));
		}
		
	}
}
